/*
 * TestAMEFProtocol.cpp
 *
 *  Created on: Mar 7, 2011
 *      Author: chhetri.sumeet
 */
#include "Timer.h"
#include "TestAMEFProtocol.h"
#include "Client.h"
#include "AMEFResources.h"
#include "PropFileReader.h"
#include "malloc_extension.h"
#include <stdio.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#define BACKLOGM 500
#define MAXEPOLLSIZE 1000
#include "map"
using namespace std;

typedef map<string,string> tablemap;
map<string, tablemap*> cached;
boost::mutex lock;

void backupTables();
void bootFromBackupTables();

int main1()
{
	Client client;
	client.connection("localhost",7001);
	AMEFEncoder* encoder = new AMEFEncoder();

	Timer t;
	AMEFObject* query = NULL;
	for (int var = 0; var < 500; ++var)
	{
		t.start();
		AMEFObject* object = new AMEFObject();
		object->addNullPacket(AMEFObject::NULL_STRING);
		object->addPacket("asdasD");
		object->addPacket(12312321);
		object->addPacket(true);
		object->addPacket(1233444555);

		query = new AMEFObject();
		query->addPacket("insert sent");
		query->addPacket(encoder->encodeWL(object, true));
		string dat = encoder->encodeB(query, false);
		client.sendData(dat);
		string ret = client.getData();
		cout << ret << " Insert " << t.elapsedMilliSeconds() << endl;
		//object->~AMEFObject();
		query->~AMEFObject();
	}

	for (int var = 0; var < 500; ++var)
	{
		t.start();

		AMEFObject* object = new AMEFObject();
		object->addPacket("aaaaaaaaa","1");

		query = new AMEFObject();
		query->addPacket("update sent 498");
		query->addPacket(encoder->encodeWL(object, false));
		string dat = encoder->encodeB(query, false);
		client.sendData(dat);
		string ret = client.getData();
		cout << ret << " Update " << t.elapsedMilliSeconds() << endl;
		//object->~AMEFObject();
		query->~AMEFObject();
	}

	//sleep(2);

	for (int var = 0; var < 500; ++var)
	{
		AMEFDecoder decoder;
		t.start();
		query = new AMEFObject();
		query->addPacket("select sent = 498");
		string dat = encoder->encodeB(query, false);
		client.sendData(dat);
		string ret = client.getData();
		AMEFObject* obj = decoder.decodeB(ret.substr(0,ret.length()-1),false,true);
		cout << "Select " << t.elapsedMilliSeconds() << endl;
		query->~AMEFObject();
	}
	encoder->~AMEFEncoder();
	client.closeConnection();
	return 0;
}

void sigchld_handler(int s)
{
    while(waitpid(-1, NULL, WNOHANG) > 0);
}

void signalSIGSEGV(int dummy)
{
	signal(SIGSEGV,signalSIGSEGV);
	void * array[25];
	int nSize = backtrace(array, 25);
	char ** symbols = backtrace_symbols(array, nSize);
	string tempo;
	for (int i = 0; i < nSize; i++)
	{
		tempo = symbols[i];
		tempo += "\n";
	}
	free(symbols);
	cout << "segmentation fault" << getpid() << "\n" << tempo << flush;
	exit(0);
}

void signalSIGFPE(int dummy)
{
	signal(SIGFPE,signalSIGFPE);
	void * array[25];
	int nSize = backtrace(array, 25);
	char ** symbols = backtrace_symbols(array, nSize);
	string tempo;
	for (int i = 0; i < nSize; i++)
	{
		tempo = symbols[i];
		tempo += "\n";
	}
	free(symbols);
	cout << "floating point exception " << getpid() << "\n" << tempo << flush;
	exit(0);
}

void signalSIGABRT(int dummy)
{
	signal(SIGABRT,signalSIGABRT);
	void * array[25];
	int nSize = backtrace(array, 25);
	char ** symbols = backtrace_symbols(array, nSize);
	string tempo;
	for (int i = 0; i < nSize; i++)
	{
		tempo = symbols[i];
		tempo += "\n";
	}
	free(symbols);
	cout << "Abort called on process" << getpid() << "\n" << tempo << flush;
	backupTables();
	exit(0);
}


void signalSIGTERM(int dummy)
{
	signal(SIGTERM,signalSIGTERM);
	void * array[25];
	int nSize = backtrace(array, 25);
	char ** symbols = backtrace_symbols(array, nSize);
	string tempo;
	for (int i = 0; i < nSize; i++)
	{
		tempo = symbols[i];
		tempo += "\n";
	}
	free(symbols);
	cout << "Termination called on process" << getpid() << "\n" << tempo << flush;
	backupTables();
	exit(0);
}


void signalSIGPIPE(int dummy)
{
	signal(SIGPIPE,signalSIGPIPE);
	void * array[25];
	int nSize = backtrace(array, 25);
	char ** symbols = backtrace_symbols(array, nSize);
	string tempo;
	for (int i = 0; i < nSize; i++)
	{
		tempo = symbols[i];
		tempo += "\n";
	}
	free(symbols);
	cout << "Broken pipe ignore it" << getpid() << "\n" << tempo << flush;
	//exit(0);
}

void signalSIGKILL(int dummy)
{
	signal(SIGKILL,signalSIGKILL);
	void * array[25];
	int nSize = backtrace(array, 25);
	char ** symbols = backtrace_symbols(array, nSize);
	string tempo;
	for (int i = 0; i < nSize; i++)
	{
		tempo = symbols[i];
		tempo += "\n";
	}
	free(symbols);
	cout << "Kill called on process with " << getpid() << "\n" << tempo << flush;
	backupTables();
	exit(0);
}

void signalSIGQUIT(int dummy)
{
	signal(SIGQUIT,signalSIGQUIT);
	void * array[25];
	int nSize = backtrace(array, 25);
	char ** symbols = backtrace_symbols(array, nSize);
	string tempo;
	for (int i = 0; i < nSize; i++)
	{
		tempo = symbols[i];
		tempo += "\n";
	}
	free(symbols);
	cout << "Quit called on process with " << getpid() << "\n" << tempo << flush;
	backupTables();
	exit(0);
}

int getLength(string header,int size)
{
	int totsize = header[size-1] & 0xff;
	for (int var = 0; var < size-1; var++)
	{
		totsize |= ((header[var] & 0xff) << (size-1-var)*8);
	}
	return totsize;
}

void insert(string quer,string data)
{
	vector<string> que;
	boost::iter_split(que, quer, boost::first_finder(" "));
	tablemap *mapo = cached[que[1]];

	string key = que[2];
	//cout << data.length() <<endl;
	mapo->erase(key);
	mapo->insert(pair<string,string>(key, data));
	//cout<< que[1] << mapo->size() <<key<<endl;
}

bool update(string quer,string data)
{
	vector<string> que;
	boost::iter_split(que, quer, boost::first_finder(" "));
	tablemap *mapo = cached[que[1]];
	if(mapo==NULL)
		return false;
	string key = que[2];
	//cout << data.length() <<endl;
	if(mapo->find(key)==mapo->end())
		return false;
	string odata = (*mapo)[key];
	AMEFDecoder decoder;

	AMEFObject* obje = decoder.decodeB(data,false,false);
	AMEFObject* objo = decoder.decodeB(odata,false,true);
	if(obje==NULL || objo==NULL)
		return false;
	AMEFObject* objn = new AMEFObject;
	for (int i = 0; i < (int)objo->getPackets().size(); i++)
	{
		bool flag = false;
		for (int j = 0; j < (int)obje->getPackets().size(); j++)
		{
			if(obje->getPackets().at(j)->getNameStr()==boost::lexical_cast<string>(i))
			{
				flag = true;
				if(obje->getPackets().at(j)->isNull())
				{
					objn->addNullPacket(obje->getPackets().at(j)->getType());
				}
				else if(obje->getPackets().at(j)->isString() || obje->getPackets().at(j)->isDate())
				{
					objn->addPacket(obje->getPackets().at(j)->getValue());
				}
				else if(obje->getPackets().at(j)->isFloatingPoint())
				{
					objn->addPacket(obje->getPackets().at(j)->getDoubleValue());
				}
				else if(obje->getPackets().at(j)->isNumber())
				{
					objn->addPacket(obje->getPackets().at(j)->getNumericValue());
				}
				else if(obje->getPackets().at(j)->getType()=='b')
				{
					objn->addPacket(obje->getPackets().at(j)->getBooleanValue());
				}
				else
				{
					objn->addPacket(obje->getPackets().at(j)->getValue()[0]);
				}
			}
		}
		if(!flag)
		{
			if(objo->getPackets().at(i)->isNull())
			{
				objn->addNullPacket(objo->getPackets().at(i)->getType());
			}
			else if(objo->getPackets().at(i)->isString() || objo->getPackets().at(i)->isDate())
			{
				objn->addPacket(objo->getPackets().at(i)->getValue());
			}
			else if(objo->getPackets().at(i)->isFloatingPoint())
			{
				objn->addPacket(objo->getPackets().at(i)->getDoubleValue());
			}
			else if(objo->getPackets().at(i)->isNumber())
			{
				objn->addPacket(objo->getPackets().at(i)->getNumericValue());
			}
			else if(objo->getPackets().at(i)->getType()=='b')
			{
				objn->addPacket(objo->getPackets().at(i)->getBooleanValue());
			}
			else
			{
				objn->addPacket(objo->getPackets().at(i)->getValue()[0]);
			}
		}
	}
	AMEFEncoder encoder;
	string ndata = encoder.encodeWL(objn, true);
	mapo->erase(key);
	mapo->insert(pair<string,string>(key, ndata));
	if(obje!=NULL)delete obje;
	if(objo!=NULL)delete objo;
	delete objn;
	//cout<< que[1] << " " << key << " " << ndata <<key<<endl;
	return true;
}

bool updatefld(string quer,string data,bool add)
{
	vector<string> que;
	boost::iter_split(que, quer, boost::first_finder(" "));
	tablemap *mapo = cached[que[1]];
	if(mapo==NULL)
		return false;
	string key = que[2];
	//cout << data.length() <<endl;
	if(mapo->find(key)==mapo->end())
		return false;
	string odata = (*mapo)[key];
	AMEFDecoder decoder;

	AMEFObject* obje = decoder.decodeB(data,false,false);
	AMEFObject* objo = decoder.decodeB(odata,false,true);

	if(obje==NULL || objo==NULL)
		return false;
	AMEFObject* objn = new AMEFObject;
	for (int i = 0; i < (int)objo->getPackets().size(); i++)
	{
		bool flag = false;
		for (int j = 0; j < (int)obje->getPackets().size(); j++)
		{
			if(obje->getPackets().at(j)->getNameStr()==boost::lexical_cast<string>(i))
			{
				flag = true;
				if(obje->getPackets().at(j)->isFloatingPoint())
				{
					if(add)
					{
						//cout << objo->getPackets().at(i)->getDoubleValue()+obje->getPackets().at(j)->getDoubleValue() << endl;
						objn->addPacket(objo->getPackets().at(i)->getDoubleValue()+obje->getPackets().at(j)->getDoubleValue());
					}
					else
					{
						//cout << objo->getPackets().at(i)->getDoubleValue()-obje->getPackets().at(j)->getDoubleValue() << endl;
						objn->addPacket(objo->getPackets().at(i)->getDoubleValue()-obje->getPackets().at(j)->getDoubleValue());
					}
				}
				else if(obje->getPackets().at(j)->isNumber())
				{
					if(add)
					{
						//cout << objo->getPackets().at(i)->getDoubleValue()+obje->getPackets().at(j)->getDoubleValue() << endl;
						objn->addPacket(objo->getPackets().at(i)->getNumericValue()+obje->getPackets().at(j)->getNumericValue());
					}
					else
					{
						//cout << objo->getPackets().at(i)->getDoubleValue()-obje->getPackets().at(j)->getDoubleValue() << endl;
						objn->addPacket(objo->getPackets().at(i)->getNumericValue()-obje->getPackets().at(j)->getNumericValue());
					}

				}
			}
		}
		if(!flag)
		{
			if(objo->getPackets().at(i)->isNull())
			{
				objn->addNullPacket(objo->getPackets().at(i)->getType());
			}
			else if(objo->getPackets().at(i)->isString() || objo->getPackets().at(i)->isDate())
			{
				objn->addPacket(objo->getPackets().at(i)->getValue());
			}
			else if(objo->getPackets().at(i)->isFloatingPoint())
			{
				objn->addPacket(objo->getPackets().at(i)->getDoubleValue());
			}
			else if(objo->getPackets().at(i)->isNumber())
			{
				objn->addPacket(objo->getPackets().at(i)->getNumericValue());
			}
			else if(objo->getPackets().at(i)->getType()=='b')
			{
				objn->addPacket(objo->getPackets().at(i)->getBooleanValue());
			}
			else
			{
				objn->addPacket(objo->getPackets().at(i)->getValue()[0]);
			}
		}
	}
	AMEFEncoder encoder;
	string ndata = encoder.encodeWL(objn, true);
	mapo->erase(key);
	mapo->insert(pair<string,string>(key, ndata));
	if(obje!=NULL)delete obje;
	if(objo!=NULL)delete objo;
	delete objn;
	//cout<< que[1] << " " << key << " " << ndata <<key<<endl;
	return true;
}


int custppchk(string quer,string data)
{
	vector<string> que;
	boost::iter_split(que, quer, boost::first_finder(" "));
	tablemap *mapo = cached[que[1]];
	if(mapo==NULL)
		return 0;
	string key = que[2];
	//cout << data.length() <<endl;
	if(mapo->find(key)==mapo->end())
		return 0;
	string odata = (*mapo)[key];
	AMEFDecoder decoder;

	AMEFObject* obje = decoder.decodeB(data,false,false);
	AMEFObject* objo = decoder.decodeB(odata,false,true);

	if(obje==NULL || objo==NULL)
		return 0;
	AMEFObject* objn = new AMEFObject;
	for (int i = 0; i < (int)objo->getPackets().size(); i++)
	{
		bool flag = false;
		for (int j = 0; j < (int)obje->getPackets().size(); j++)
		{
			if(obje->getPackets().at(j)->getNameStr()==boost::lexical_cast<string>(i))
			{
				flag = true;
				if(obje->getPackets().at(j)->isFloatingPoint())
				{
					if(objo->getPackets().at(i)->getDoubleValue()<obje->getPackets().at(j)->getDoubleValue())
						return 2;
					else
						objn->addPacket(objo->getPackets().at(i)->getDoubleValue()-obje->getPackets().at(j)->getDoubleValue());
				}
				else if(obje->getPackets().at(j)->isNumber())
				{
					if(objo->getPackets().at(i)->getDoubleValue()<obje->getPackets().at(j)->getDoubleValue())
						return 2;
					else
						objn->addPacket(objo->getPackets().at(i)->getNumericValue()-obje->getPackets().at(j)->getNumericValue());
				}
			}
		}
		if(!flag)
		{
			if(objo->getPackets().at(i)->isNull())
			{
				objn->addNullPacket(objo->getPackets().at(i)->getType());
			}
			else if(objo->getPackets().at(i)->isString() || objo->getPackets().at(i)->isDate())
			{
				objn->addPacket(objo->getPackets().at(i)->getValue());
			}
			else if(objo->getPackets().at(i)->isFloatingPoint())
			{
				objn->addPacket(objo->getPackets().at(i)->getDoubleValue());
			}
			else if(objo->getPackets().at(i)->isNumber())
			{
				objn->addPacket(objo->getPackets().at(i)->getNumericValue());
			}
			else if(objo->getPackets().at(i)->getType()=='b')
			{
				objn->addPacket(objo->getPackets().at(i)->getBooleanValue());
			}
			else
			{
				objn->addPacket(objo->getPackets().at(i)->getValue()[0]);
			}
		}
	}
	AMEFEncoder encoder;
	string ndata = encoder.encodeWL(objn, true);
	mapo->erase(key);
	mapo->insert(pair<string,string>(key, ndata));
	if(obje!=NULL)delete obje;
	if(objo!=NULL)delete objo;
	delete objn;
	//cout<< que[1] << " " << key << " " << ndata <<key<<endl;
	return 1;
}


bool updmov(string quer,string val,bool flag)
{
	AMEFDecoder decoder;
	vector<string> que;
	boost::iter_split(que, quer, boost::first_finder(" "));
	tablemap *mapof = cached[que[1]];
	tablemap *mapot = cached[que[2]];
	if(mapof==NULL || mapot==NULL)
		return false;
	string keyo = que[3];
	string key = keyo;
	if(que.size()==5)
		key = que[4];
	AMEFObject* objn = new AMEFObject;
	if(mapof->find(keyo)==mapof->end())
		return false;
	AMEFObject* objo = decoder.decodeB((*mapof)[keyo],false,true);
	if(objo==NULL)
		return false;
	AMEFObject* obje = decoder.decodeB(val,false,false);
	if(obje==NULL)
		return false;
	for (int i = 0; i < (int)objo->getPackets().size(); i++)
	{
		bool flag = false;
		for (int j = 0; j < (int)obje->getPackets().size(); j++)
		{
			if(obje->getPackets().at(j)->getNameStr()==boost::lexical_cast<string>(i))
			{
				flag = true;
				if(obje->getPackets().at(j)->isNull())
				{
					objn->addNullPacket(obje->getPackets().at(j)->getType());
				}
				else if(obje->getPackets().at(j)->isString() || obje->getPackets().at(j)->isDate())
				{
					objn->addPacket(obje->getPackets().at(j)->getValue());
				}
				else if(obje->getPackets().at(j)->isFloatingPoint())
				{
					objn->addPacket(obje->getPackets().at(j)->getDoubleValue());
				}
				else if(obje->getPackets().at(j)->isNumber())
				{
					objn->addPacket(obje->getPackets().at(j)->getNumericValue());
				}
				else if(obje->getPackets().at(j)->getType()=='b')
				{
					objn->addPacket(obje->getPackets().at(j)->getBooleanValue());
				}
				else
				{
					objn->addPacket(obje->getPackets().at(j)->getValue()[0]);
				}
			}
		}
		if(!flag)
		{
			if(objo->getPackets().at(i)->isNull())
			{
				objn->addNullPacket(objo->getPackets().at(i)->getType());
			}
			else if(objo->getPackets().at(i)->isString() || objo->getPackets().at(i)->isDate())
			{
				objn->addPacket(objo->getPackets().at(i)->getValue());
			}
			else if(objo->getPackets().at(i)->isFloatingPoint())
			{
				objn->addPacket(objo->getPackets().at(i)->getDoubleValue());
			}
			else if(objo->getPackets().at(i)->isNumber())
			{
				objn->addPacket(objo->getPackets().at(i)->getNumericValue());
			}
			else if(objo->getPackets().at(i)->getType()=='b')
			{
				objn->addPacket(objo->getPackets().at(i)->getBooleanValue());
			}
			else
			{
				objn->addPacket(objo->getPackets().at(i)->getValue()[0]);
			}
		}
	}
	AMEFEncoder encoder;
	string ndata = encoder.encodeWL(objn, true);
	lock.lock();
	if(flag)mapof->erase(keyo);
	mapot->insert(pair<string,string>(key, ndata));
	lock.unlock();
	if(obje!=NULL)delete obje;
	if(objo!=NULL)delete objo;
	delete objn;
	return true;
	//cout<< que[1]  <<":"<< keyo <<":"<<mapof->size()<< "  " << que[2] <<":"<<key<<":"<<mapot->size()<<endl;
}

bool upmovwrtpvap(string quer,string val,bool flag)
{
	AMEFDecoder decoder;
	vector<string> que;
	boost::iter_split(que, quer, boost::first_finder(" "));
	tablemap *mapof = cached[que[1]];
	string pre = que[2];
	if(mapof==NULL)
		return false;
	string keyo = que[3];
	string key = que[4];
	string ind = que[5];
	tablemap *mapot = NULL;
	if(mapof->find(keyo)==mapof->end())
			return false;
	AMEFObject* objo = decoder.decodeB((*mapof)[keyo],false,true);
	if(objo!=NULL)
	{
		string post = objo->getPackets().at(boost::lexical_cast<int>(ind))->getValueStr();
		mapot = cached[pre+post];
		if(mapot==NULL)
		{
			delete objo;
			return false;
		}
		if(objo->getPackets().size()>boost::lexical_cast<long>(key))
		{
			key = objo->getPackets().at(boost::lexical_cast<long>(key))->getValueStr();
		}
		else
		{
			delete objo;
			return false;
		}
	}
	else
		return false;
	AMEFObject* objn = new AMEFObject;
	AMEFObject* obje = decoder.decodeB(val,false,false);
	if(obje==NULL)
	{
		delete objn;
		return false;
	}
	for (int i = 0; i < (int)objo->getPackets().size(); i++)
	{
		bool flag = false;
		for (int j = 0; j < (int)obje->getPackets().size(); j++)
		{
			if(obje->getPackets().at(j)->getNameStr()==boost::lexical_cast<string>(i))
			{
				flag = true;
				if(obje->getPackets().at(j)->isNull())
				{
					objn->addNullPacket(obje->getPackets().at(j)->getType());
				}
				else if(obje->getPackets().at(j)->isString() || obje->getPackets().at(j)->isDate())
				{
					objn->addPacket(obje->getPackets().at(j)->getValue());
				}
				else if(obje->getPackets().at(j)->isFloatingPoint())
				{
					objn->addPacket(obje->getPackets().at(j)->getDoubleValue());
				}
				else if(obje->getPackets().at(j)->isNumber())
				{
					objn->addPacket(obje->getPackets().at(j)->getNumericValue());
				}
				else if(obje->getPackets().at(j)->getType()=='b')
				{
					objn->addPacket(obje->getPackets().at(j)->getBooleanValue());
				}
				else
				{
					objn->addPacket(obje->getPackets().at(j)->getValue()[0]);
				}
			}
		}
		if(!flag)
		{
			if(objo->getPackets().at(i)->isNull())
			{
				objn->addNullPacket(objo->getPackets().at(i)->getType());
			}
			else if(objo->getPackets().at(i)->isString() || objo->getPackets().at(i)->isDate())
			{
				objn->addPacket(objo->getPackets().at(i)->getValue());
			}
			else if(objo->getPackets().at(i)->isFloatingPoint())
			{
				objn->addPacket(objo->getPackets().at(i)->getDoubleValue());
			}
			else if(objo->getPackets().at(i)->isNumber())
			{
				objn->addPacket(objo->getPackets().at(i)->getNumericValue());
			}
			else if(objo->getPackets().at(i)->getType()=='b')
			{
				objn->addPacket(objo->getPackets().at(i)->getBooleanValue());
			}
			else
			{
				objn->addPacket(objo->getPackets().at(i)->getValue()[0]);
			}
		}
	}
	AMEFEncoder encoder;
	string ndata = encoder.encodeWL(objn, true);
	lock.lock();
	if(flag)mapof->erase(keyo);
	mapot->insert(pair<string,string>(key, ndata));
	lock.unlock();
	delete objo;
	delete obje;
	delete objn;
	return true;
	//cout<< que[1]  <<":"<< keyo <<":"<<mapof->size()<< "  " << que[2] <<":"<<key<<":"<<mapot->size()<<endl;
}

bool upmovwrt(string quer,string val,bool flag)
{
	AMEFDecoder decoder;
	vector<string> que;
	boost::iter_split(que, quer, boost::first_finder(" "));
	tablemap *mapof = cached[que[1]];
	string pre = que[2];
	if(mapof==NULL)
		return false;
	string keyo = que[3];
	string key = que[4];
	string ind = que[5];
	tablemap *mapot = NULL;
	if(mapof->find(keyo)==mapof->end())
			return false;
	AMEFObject* objo = decoder.decodeB((*mapof)[keyo],false,true);
	if(objo!=NULL)
	{
		string post = objo->getPackets().at(boost::lexical_cast<int>(ind))->getValueStr();
		mapot = cached[pre+post];
		if(mapot==NULL)
		{
			delete objo;
			return false;
		}
	}
	else
		return false;
	AMEFObject* objn = new AMEFObject;
	AMEFObject* obje = decoder.decodeB(val,false,false);
	if(obje==NULL)
	{
		delete objn;
		return false;
	}
	for (int i = 0; i < (int)objo->getPackets().size(); i++)
	{
		bool flag = false;
		for (int j = 0; j < (int)obje->getPackets().size(); j++)
		{
			if(obje->getPackets().at(j)->getNameStr()==boost::lexical_cast<string>(i))
			{
				flag = true;
				if(obje->getPackets().at(j)->isNull())
				{
					objn->addNullPacket(obje->getPackets().at(j)->getType());
				}
				else if(obje->getPackets().at(j)->isString() || obje->getPackets().at(j)->isDate())
				{
					objn->addPacket(obje->getPackets().at(j)->getValue());
				}
				else if(obje->getPackets().at(j)->isFloatingPoint())
				{
					objn->addPacket(obje->getPackets().at(j)->getDoubleValue());
				}
				else if(obje->getPackets().at(j)->isNumber())
				{
					objn->addPacket(obje->getPackets().at(j)->getNumericValue());
				}
				else if(obje->getPackets().at(j)->getType()=='b')
				{
					objn->addPacket(obje->getPackets().at(j)->getBooleanValue());
				}
				else
				{
					objn->addPacket(obje->getPackets().at(j)->getValue()[0]);
				}
			}
		}
		if(!flag)
		{
			if(objo->getPackets().at(i)->isNull())
			{
				objn->addNullPacket(objo->getPackets().at(i)->getType());
			}
			else if(objo->getPackets().at(i)->isString() || objo->getPackets().at(i)->isDate())
			{
				objn->addPacket(objo->getPackets().at(i)->getValue());
			}
			else if(objo->getPackets().at(i)->isFloatingPoint())
			{
				objn->addPacket(objo->getPackets().at(i)->getDoubleValue());
			}
			else if(objo->getPackets().at(i)->isNumber())
			{
				objn->addPacket(objo->getPackets().at(i)->getNumericValue());
			}
			else if(objo->getPackets().at(i)->getType()=='b')
			{
				objn->addPacket(objo->getPackets().at(i)->getBooleanValue());
			}
			else
			{
				objn->addPacket(objo->getPackets().at(i)->getValue()[0]);
			}
		}
	}
	AMEFEncoder encoder;
	string ndata = encoder.encodeWL(objn, true);
	lock.lock();
	if(flag)mapof->erase(keyo);
	mapot->insert(pair<string,string>(key, ndata));
	lock.unlock();
	delete objo;
	delete obje;
	delete objn;
	return true;
	//cout<< que[1]  <<":"<< keyo <<":"<<mapof->size()<< "  " << que[2] <<":"<<key<<":"<<mapot->size()<<endl;
}

int main4()
{
	for(int i = 1; i < 10; ++i){
		cached["sent"] = new map<string,string>;
		for (int var = 1; var < 10000; ++var) {
			string dat = "asdasdasdasdasdasdas";
			AMEFObject* object = new AMEFObject();
			object->addNullPacket(AMEFObject::NULL_STRING);
			object->addPacket("asdasD");
			object->addPacket(12312321);
			object->addPacket(true);
			object->addPacket(1233444555);
			cached["sent"]->insert(pair<string,string>(boost::lexical_cast<string>(var), "asasd"));
			delete object;
		}
		cout << "done writing" << endl;
		sleep(5);
		cached["sent"]->clear();
		cached["sent"]->empty();
		cout << "done clearing" << endl;
		sleep(5);
	}
}


bool move(string quer,bool flag)
{
	vector<string> que;
	boost::iter_split(que, quer, boost::first_finder(" "));
	tablemap *mapof = cached[que[1]];
	tablemap *mapot = cached[que[2]];
	if(mapof==NULL || mapot==NULL)
		return false;
	string keyo = que[3];
	string key = keyo;
	if(que.size()==5)
		key = que[4];
	////cout << data.length() <<endl;
	lock.lock();
	string dat = (*mapof)[keyo];
	if(flag)mapof->erase(keyo);
	mapot->insert(pair<string,string>(key, dat));
	lock.unlock();
	//cout<< que[1]  <<":"<< keyo <<":"<<mapof->size()<< "  " << que[2] <<":"<<key<<":"<<mapot->size()<<endl;
	return true;
}

bool movepvap(string quer,bool flag)
{
	AMEFDecoder decoder;
	vector<string> que;
	boost::iter_split(que, quer, boost::first_finder(" "));
	tablemap *mapof = cached[que[1]];
	string pre = que[2];
	if(mapof==NULL)
		return false;
	string keyo = que[3];
	string key = que[4];
	string ind = que[5];
	if(mapof->find(keyo)==mapof->end())
		return false;
	AMEFObject* obj = decoder.decodeB((*mapof)[keyo],false,true);
	if(obj==NULL)
		return false;
	string post = obj->getPackets().at(boost::lexical_cast<int>(ind))->getValueStr();
	delete obj;
	tablemap *mapot = cached[pre+post];
	if(mapot==NULL)
		return false;
	////cout << data.length() <<endl;
	lock.lock();
	string dat = (*mapof)[keyo];
	if(flag)mapof->erase(keyo);
	mapot->insert(pair<string,string>(key, dat));
	lock.unlock();
	//cout<< que[1]  <<":"<< keyo <<":"<<mapof->size()<< "  " << que[2] <<":"<<key<<":"<<mapot->size()<<endl;
	return true;
}

bool movewrtpvap(string quer,bool flag)
{
	AMEFDecoder decoder;
	vector<string> que;
	boost::iter_split(que, quer, boost::first_finder(" "));
	tablemap *mapof = cached[que[1]];
	string pre = que[2];
	if(mapof==NULL)
		return false;
	string keyo = que[3];
	string key = que[4];
	string ind = que[5];
	if(mapof->find(keyo)==mapof->end())
		return false;
	AMEFObject* obj = decoder.decodeB((*mapof)[keyo],false,true);
	if(obj!=NULL)
	{
		string post = obj->getPackets().at(boost::lexical_cast<int>(ind))->getValueStr();
		tablemap *mapot = cached[pre+post];
		//cout << pre+post << " " << mapot << endl;
		if(mapot==NULL)
			return false;
		if(obj->getPackets().size()>boost::lexical_cast<long>(key))
		{
			key = obj->getPackets().at(boost::lexical_cast<long>(key))->getValueStr();
		}
		//cout << key << endl;
		lock.lock();
		string dat = (*mapof)[keyo];
		if(flag)mapof->erase(keyo);
		mapot->insert(pair<string,string>(key, dat));
		lock.unlock();
		delete obj;
	}
	//cout << "movewrt command key is "<< key <<endl;
	return true;
	//cout<< que[1]  <<":"<< keyo <<":"<<mapof->size()<< "  " << que[2] <<":"<<key<<":"<<mapot->size()<<endl;
}


bool movewrt(string quer,bool flag)
{
	AMEFDecoder decoder;
	vector<string> que;
	boost::iter_split(que, quer, boost::first_finder(" "));
	tablemap *mapof = cached[que[1]];
	tablemap *mapot = cached[que[2]];
	if(mapof==NULL || mapot==NULL)
		return false;
	string keyo = que[3];
	string key = keyo;
	if(que.size()==5)
		key = que[4];
	if(mapof->find(keyo)==mapof->end())
		return false;
	AMEFObject* obj = decoder.decodeB((*mapof)[keyo],false,true);
	if(obj==NULL)
		return false;
	if(obj->getPackets().size()>boost::lexical_cast<long>(key))
	{
		key = obj->getPackets().at(boost::lexical_cast<long>(key))->getValueStr();
	}
	else
	{
		delete obj;
		return false;
	}
	//cout << "movewrt command key is "<< key <<endl;
	lock.lock();
	string dat = (*mapof)[keyo];
	if(flag)mapof->erase(keyo);
	mapot->insert(pair<string,string>(key, dat));
	lock.unlock();
	delete obj;
	return true;
	//cout<< que[1]  <<":"<< keyo <<":"<<mapof->size()<< "  " << que[2] <<":"<<key<<":"<<mapot->size()<<endl;
}

bool apmov(string quer,string val,bool flag)
{
	AMEFDecoder decoder;
	vector<string> que;
	boost::iter_split(que, quer, boost::first_finder(" "));
	tablemap *mapof = cached[que[1]];
	tablemap *mapot = cached[que[2]];
	if(mapof==NULL || mapot==NULL)
		return false;
	string keyo = que[3];
	string key = keyo;
	if(que.size()==5)
		key = que[4];
	AMEFObject* objn = new AMEFObject;
	if(mapof->find(keyo)==mapof->end())
		return false;
	AMEFObject* obj = decoder.decodeB((*mapof)[keyo],false,true);
	if(obj==NULL)
		return false;
	AMEFObject* obje = decoder.decodeB(val,false,true);
	if(obje==NULL)
		return false;
	for (int j = 0; j < (int)obj->getPackets().size(); j++)
	{
		if(obj->getPackets().at(j)->isNull())
		{
			objn->addNullPacket(obj->getPackets().at(j)->getType());
		}
		else if(obj->getPackets().at(j)->isString() || obj->getPackets().at(j)->isDate())
		{
			objn->addPacket(obj->getPackets().at(j)->getValue());
		}
		else if(obj->getPackets().at(j)->isFloatingPoint())
		{
			objn->addPacket(obj->getPackets().at(j)->getDoubleValue());
		}
		else if(obj->getPackets().at(j)->isNumber())
		{
			objn->addPacket(obj->getPackets().at(j)->getNumericValue());
		}
		else if(obj->getPackets().at(j)->getType()=='b')
		{
			objn->addPacket(obj->getPackets().at(j)->getBooleanValue());
		}
		else
		{
			objn->addPacket(obj->getPackets().at(j)->getValue()[0]);
		}
	}
	for (int j = 0; j < (int)obje->getPackets().size(); j++)
	{
		if(obje->getPackets().at(j)->isNull())
		{
			objn->addNullPacket(obje->getPackets().at(j)->getType());
		}
		else if(obje->getPackets().at(j)->isString() || obje->getPackets().at(j)->isDate())
		{
			objn->addPacket(obje->getPackets().at(j)->getValue());
		}
		else if(obje->getPackets().at(j)->isFloatingPoint())
		{
			objn->addPacket(obje->getPackets().at(j)->getDoubleValue());
		}
		else if(obje->getPackets().at(j)->isNumber())
		{
			objn->addPacket(obje->getPackets().at(j)->getNumericValue());
		}
		else if(obje->getPackets().at(j)->getType()=='b')
		{
			objn->addPacket(obje->getPackets().at(j)->getBooleanValue());
		}
		else
		{
			objn->addPacket(obje->getPackets().at(j)->getValue()[0]);
		}
	}
	AMEFEncoder encoder;
	string ndata = encoder.encodeWL(objn, true);
	lock.lock();
	if(flag)mapof->erase(keyo);
	mapot->insert(pair<string,string>(key, ndata));
	lock.unlock();
	if(obje!=NULL)delete obje;
	if(obj!=NULL)delete obj;
	delete objn;
	return true;
	//cout<< que[1]  <<":"<< keyo <<":"<<mapof->size()<< "  " << que[2] <<":"<<key<<":"<<mapot->size()<<endl;
}


bool apmovwrt(string quer,string val,bool flag)
{
	AMEFDecoder decoder;
	vector<string> que;
	boost::iter_split(que, quer, boost::first_finder(" "));
	tablemap *mapof = cached[que[1]];
	tablemap *mapot = cached[que[2]];
	if(mapof==NULL || mapot==NULL)
		return false;
	string keyo = que[3];
	string key = keyo;
	if(que.size()==5)
		key = que[4];
	if(mapof->find(keyo)==mapof->end())
		return false;
	AMEFObject* obj = decoder.decodeB((*mapof)[keyo],false,true);
	if(obj==NULL)
	{
		return false;
	}
	if(obj->getPackets().size()>boost::lexical_cast<long>(key))
	{
		key = obj->getPackets().at(boost::lexical_cast<long>(key))->getValueStr();
	}
	else
	{
		delete obj;
		return false;
	}
	AMEFObject* objn = new AMEFObject;
	AMEFObject* obje = decoder.decodeB(val,false,true);
	if(obje==NULL)
	{
		delete objn;
		return false;
	}
	for (int j = 0; j < (int)obj->getPackets().size(); j++)
	{
		if(obj->getPackets().at(j)->isNull())
		{
			objn->addNullPacket(obj->getPackets().at(j)->getType());
		}
		else if(obj->getPackets().at(j)->isString() || obj->getPackets().at(j)->isDate())
		{
			objn->addPacket(obj->getPackets().at(j)->getValue());
		}
		else if(obj->getPackets().at(j)->isFloatingPoint())
		{
			objn->addPacket(obj->getPackets().at(j)->getDoubleValue());
		}
		else if(obj->getPackets().at(j)->isNumber())
		{
			objn->addPacket(obj->getPackets().at(j)->getNumericValue());
		}
		else if(obj->getPackets().at(j)->getType()=='b')
		{
			objn->addPacket(obj->getPackets().at(j)->getBooleanValue());
		}
		else
		{
			objn->addPacket(obj->getPackets().at(j)->getValue()[0]);
		}
	}
	for (int j = 0; j < (int)obje->getPackets().size(); j++)
	{
		if(obje->getPackets().at(j)->isNull())
		{
			objn->addNullPacket(obje->getPackets().at(j)->getType());
		}
		else if(obje->getPackets().at(j)->isString() || obje->getPackets().at(j)->isDate())
		{
			objn->addPacket(obje->getPackets().at(j)->getValue());
		}
		else if(obje->getPackets().at(j)->isFloatingPoint())
		{
			objn->addPacket(obje->getPackets().at(j)->getDoubleValue());
		}
		else if(obje->getPackets().at(j)->isNumber())
		{
			objn->addPacket(obje->getPackets().at(j)->getNumericValue());
		}
		else if(obje->getPackets().at(j)->getType()=='b')
		{
			objn->addPacket(obje->getPackets().at(j)->getBooleanValue());
		}
		else
		{
			objn->addPacket(obje->getPackets().at(j)->getValue()[0]);
		}
	}
	AMEFEncoder encoder;
	string ndata = encoder.encodeWL(objn, true);
	lock.lock();
	if(flag)mapof->erase(keyo);
	mapot->insert(pair<string,string>(key, ndata));
	lock.unlock();
	delete obj;
	delete obje;
	delete objn;
	return true;
	//cout<< que[1]  <<":"<< keyo <<":"<<mapof->size()<< "  " << que[2] <<":"<<key<<":"<<mapot->size()<<endl;
}



bool apmovwrtpvap(string quer,string val,bool flag)
{
	AMEFDecoder decoder;
	vector<string> que;
	boost::iter_split(que, quer, boost::first_finder(" "));
	tablemap *mapof = cached[que[1]];
	string pre = que[2];
	if(mapof==NULL)
		return false;
	string keyo = que[3];
	string key = que[4];
	string ind = que[5];
	tablemap *mapot = NULL;
	if(mapof->find(keyo)==mapof->end())
			return false;
	AMEFObject* obj = decoder.decodeB((*mapof)[keyo],false,true);
	if(obj!=NULL)
	{
		string post = obj->getPackets().at(boost::lexical_cast<int>(ind))->getValueStr();
		mapot = cached[pre+post];
		if(mapot==NULL)
		{
			delete obj;
			return false;
		}
		if(obj->getPackets().size()>boost::lexical_cast<long>(key))
		{
			key = obj->getPackets().at(boost::lexical_cast<long>(key))->getValueStr();
		}
		else
		{
			delete obj;
			return false;
		}
	}
	else
		return false;
	AMEFObject* objn = new AMEFObject;
	AMEFObject* obje = decoder.decodeB(val,false,true);
	if(obje==NULL)
	{
		delete objn;
		return false;
	}
	for (int j = 0; j < (int)obj->getPackets().size(); j++)
	{
		if(obj->getPackets().at(j)->isNull())
		{
			objn->addNullPacket(obj->getPackets().at(j)->getType());
		}
		else if(obj->getPackets().at(j)->isString() || obj->getPackets().at(j)->isDate())
		{
			objn->addPacket(obj->getPackets().at(j)->getValue());
		}
		else if(obj->getPackets().at(j)->isFloatingPoint())
		{
			objn->addPacket(obj->getPackets().at(j)->getDoubleValue());
		}
		else if(obj->getPackets().at(j)->isNumber())
		{
			objn->addPacket(obj->getPackets().at(j)->getNumericValue());
		}
		else if(obj->getPackets().at(j)->getType()=='b')
		{
			objn->addPacket(obj->getPackets().at(j)->getBooleanValue());
		}
		else
		{
			objn->addPacket(obj->getPackets().at(j)->getValue()[0]);
		}
	}
	for (int j = 0; j < (int)obje->getPackets().size(); j++)
	{
		if(obje->getPackets().at(j)->isNull())
		{
			objn->addNullPacket(obje->getPackets().at(j)->getType());
		}
		else if(obje->getPackets().at(j)->isString() || obje->getPackets().at(j)->isDate())
		{
			objn->addPacket(obje->getPackets().at(j)->getValue());
		}
		else if(obje->getPackets().at(j)->isFloatingPoint())
		{
			objn->addPacket(obje->getPackets().at(j)->getDoubleValue());
		}
		else if(obje->getPackets().at(j)->isNumber())
		{
			objn->addPacket(obje->getPackets().at(j)->getNumericValue());
		}
		else if(obje->getPackets().at(j)->getType()=='b')
		{
			objn->addPacket(obje->getPackets().at(j)->getBooleanValue());
		}
		else
		{
			objn->addPacket(obje->getPackets().at(j)->getValue()[0]);
		}
	}
	AMEFEncoder encoder;
	string ndata = encoder.encodeWL(objn, true);
	lock.lock();
	if(flag)mapof->erase(keyo);
	mapot->insert(pair<string,string>(key, ndata));
	lock.unlock();
	delete obj;
	delete obje;
	delete objn;
	return true;
	//cout<< que[1]  <<":"<< keyo <<":"<<mapof->size()<< "  " << que[2] <<":"<<key<<":"<<mapot->size()<<endl;
}

int main222(int argc, char* argv[])
{
	if("476721234568">"0")cout << "success"<<endl;
}

void static getStatsAndClear()
{
	string stats;
	ifstream ifs("serv.ctrl");
	while(ifs.is_open())
	{
		ifs.close();
		ifs.open("serv.ctrl");
		sleep(60);
		char buf[1000];
		//MallocExtension::instance()->GetStats(buf,1000);
		MallocExtension::instance()->ReleaseFreeMemory();
		cout << "released memory back to OS" << endl;
	}
}

int main(int argc, char* argv[])
{
	PropFileReader prop;
	propMap props = prop.getProperties("gcache.prop");

	string port = props["port"];
	string tables = props["tables"];

	strVec ipprts;
	boost::iter_split(ipprts, tables, boost::first_finder(","));
	for (int var = 0; var < (int)ipprts.size(); var++)
	{
		cached[ipprts.at(var)] = new map<string,string>;
	}
	bootFromBackupTables();

	/*cached["send"] = new map<string,string>;
	cached["sendhlr"] = new map<string,string>;
	cached["deliver"] = new map<string,string>;
	cached["delivered"] = new map<string,string>;
	cached["deliverednodlr"] = new map<string,string>;
	cached["routes"] = new map<string,string>;
	cached["provision"] = new map<string,string>;
	cached["client"] = new map<string,string>;
	cached["tinvalid"] = new map<string,string>;
	cached["invalid"] = new map<string,string>;
	cached["retry"] = new map<string,string>;*/
	signal(SIGSEGV,signalSIGSEGV);
	signal(SIGPIPE,signalSIGPIPE);
	signal(SIGKILL,signalSIGKILL);
	signal(SIGTERM,signalSIGTERM);
	signal(SIGABRT,signalSIGABRT);
	signal(SIGFPE,signalSIGFPE);
	signal(SIGQUIT,signalSIGQUIT);
	int sockfd, new_fd;  // listen on sock_fd, new connection on new_fd
	struct addrinfo hints, *servinfo, *p;
	struct sockaddr_storage their_addr; // connector's address information
	socklen_t sin_size;
	struct sigaction sa;


	struct epoll_event ev;
	//struct rlimit rt;
	int yes=1;
	//char s[INET6_ADDRSTRLEN];
	int rv,nfds;
	memset(&hints, 0, sizeof hints);
    hints.ai_family = AF_UNSPEC;
    hints.ai_socktype = SOCK_STREAM;
    hints.ai_flags = AI_PASSIVE; // use my IP
    if ((rv = getaddrinfo(NULL, port.c_str(), &hints, &servinfo)) != 0) {
        fprintf(stderr, "getaddrinfo: %s\n", gai_strerror(rv));
        return 1;
    }

    // loop through all the results and bind to the first we can
    for(p = servinfo; p != NULL; p = p->ai_next) {
        if ((sockfd = socket(p->ai_family, p->ai_socktype,
                p->ai_protocol)) == -1) {
            perror("server: socket");
            continue;
        }
        if (setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &yes,
                sizeof(int)) == -1) {
            perror("setsockopt");
            exit(1);
        }
        if (bind(sockfd, p->ai_addr, p->ai_addrlen) == -1) {
			close(sockfd);
            perror("server: bind");
            continue;
        }
        break;
    }

    if (p == NULL)  {
        fprintf(stderr, "server: failed to bind\n");
        return 2;
    }
    freeaddrinfo(servinfo); // all done with this structure
    if (listen(sockfd, BACKLOGM) == -1) {
        perror("listen");
        exit(1);
    }
    fcntl(sockfd, F_SETFL, fcntl(sockfd, F_GETFD, 0) | O_NONBLOCK);
    sa.sa_handler = sigchld_handler; // reap all dead processes
    sigemptyset(&sa.sa_mask);
    sa.sa_flags = SA_RESTART;
    if (sigaction(SIGCHLD, &sa, NULL) == -1) {
        perror("sigaction");
        exit(1);
    }
    struct epoll_event events[MAXEPOLLSIZE];
	int epoll_handle = epoll_create(MAXEPOLLSIZE);
	ev.events = EPOLLIN | EPOLLPRI;
	ev.data.fd = sockfd;
	if (epoll_ctl(epoll_handle, EPOLL_CTL_ADD, sockfd, &ev) < 0)
	{
		fprintf(stderr, "epoll set insertion error: fd=%d\n", sockfd);
		return -1;
	}
	else
		printf("listener socket to join epoll success!\n");

	boost::thread(boost::bind(&getStatsAndClear));

    ofstream ofs("serv.ctrl");
    ofs << "Proces" << flush;
    ofs.close();
    cout << "listening on port "<< port << endl;

    ifstream ifs("serv.ctrl");
    int curfds = 1;
    char buf[4];
    while(ifs.is_open())
	{

    	nfds = epoll_wait(epoll_handle, events, curfds, 1000);
    	ifs.close();
    	ifs.open("serv.ctrl");

		if (nfds == -1)
		{
			perror("epoll_wait main process");
			//logfile << "Interruption Signal Received\n" << flush;
			printf("Interruption Signal Received\n");
			curfds = 1;
			if(errno==EBADF)
				cout << "\nInavlid fd" <<flush;
			else if(errno==EFAULT)
				cout << "\nThe memory area pointed to by events is not accessible" <<flush;
			else if(errno==EINTR)
				cout << "\ncall was interrupted by a signal handler before any of the requested events occurred" <<flush;
			else
				cout << "\nnot an epoll file descriptor" <<flush;
			//break;
		}
		vector<string> que;
		string ln,data,quer;
		for(int n=0;n<nfds;n++)
		{
			if (events[n].data.fd == sockfd)
			{
				new_fd = -1;
				sin_size = sizeof their_addr;
				new_fd = accept(sockfd, (struct sockaddr *)&their_addr, &sin_size);
				if (new_fd == -1)
				{
					perror("accept");
					continue;
				}
				else
				{
					curfds++;
					fcntl(new_fd, F_SETFL, fcntl(new_fd, F_GETFD, 0) | O_NONBLOCK);
					ev.events = EPOLLIN | EPOLLPRI;
					ev.data.fd = new_fd;
					if (epoll_ctl(epoll_handle, EPOLL_CTL_ADD, new_fd, &ev) < 0)
					{
						perror("epoll");
						cout << "\nerror adding to epoll cntl list" << flush;
						return -1;
					}
				}
			}
			else
			{
				int err;
				if((err=recv(events[n].data.fd,buf,4,MSG_WAITALL))==0)
				{
					close(events[n].data.fd);
					//cout << "\nsocket conn closed before being serviced" << flush;
					epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
					curfds--;
				}
				else if(err==4)
				{
					int fd = events[n].data.fd;
					AMEFDecoder decoder;
					//cout << "got some query" << endl;
					for (int var = 0; var < 4; var++) {
						ln.push_back(buf[var]);
					}
					int len = getLength(ln,4);
					//cout << len << endl;
					if(len>10000000)
					{
						close(events[n].data.fd);
						cout << "WARNING>> Invalid data or someone trying to attack the cache or maybe a simple telnet attempt...ignore..." << endl;
						epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						continue;
					}
					char* buff = new char[len];

					err = recv(events[n].data.fd,buff,len,MSG_WAITALL);
					if(err==0)
					{
						close(events[n].data.fd);
						//cout << "\nsocket conn closed before being serviced" << flush;
						epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						continue;
					}
					for (int var = 0; var < len; var++) {
						data.push_back(buff[var]);
					}
					if(data.find("select")==string::npos && data.find("count")==string::npos)
						cout << data << endl;
					//memset(&buf[0], 0, sizeof(buf));
					delete[] buff;
					AMEFObject* query = decoder.decodeB(data,false,false);
					//cout << data << endl;
					string quer = query->getPackets().at(0)->getValue();
					if(quer.find("insert ")!=-1 && query->getPackets().size()==2/*
										&& query->getPackets().at(1)->getValue()!=""*/)
					{
						//insert(quer, query->getPackets().at(1)->getValue());
						boost::iter_split(que, quer, boost::first_finder(" "));
						tablemap *mapo = cached[que[1]];
						if(mapo!=NULL)
						{
							string key = que[2];
							string val = query->getPackets().at(1)->getValue();
							//cout << val.length() <<endl;
							lock.lock();
							mapo->erase(key);
							mapo->insert(pair<string,string>(key, val));
							lock.unlock();
							int err = send(fd, "T", 1, 0);
							if(err<=0)
							{
								close(events[n].data.fd);
								//cout << "\nsocket conn closed before being serviced" << flush;
								epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
							}
						}
						else
						{
							int err = send(fd, "F", 1, 0);
							if(err<=0)
							{
								close(events[n].data.fd);
								//cout << "\nsocket conn closed before being serviced" << flush;
								epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
							}
						}
					}
					else if(quer.find("custppchk ")!=-1)
					{
						string val = query->getPackets().at(1)->getValue();
						int fl = custppchk(quer,val);
						AMEFObject objn;
						objn.addPacket(fl);
						AMEFEncoder encoder;
						string ndata = encoder.encodeWL(&objn, true);
						ndata += "F";
						int err = send(fd, ndata.c_str(), ndata.length(), 0);
						if(err<=0)
						{
							close(events[n].data.fd);
							//cout << "\nsocket conn closed before being serviced" << flush;
							epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						}
					}
					else if(quer.find("count ")!=-1)
					{
						boost::iter_split(que, quer, boost::first_finder(" "));
						tablemap *mapo = cached[que[1]];
						if(que.size()==2)
						{
							if(mapo!=NULL)
							{
								AMEFObject objn;
								objn.addPacket((long)mapo->size());
								AMEFEncoder encoder;
								string ndata = encoder.encodeWL(&objn, true);
								ndata += "F";
								////cout << dat << endl;
								int err = send(fd, ndata.c_str(), ndata.length(), 0);
								if(err<=0)
								{
									close(events[n].data.fd);
									//cout << "\nsocket conn closed before being serviced" << flush;
									epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
								}
							}
							else
							{
								int err = send(fd, "F", 1, 0);
								if(err<=0)
								{
									close(events[n].data.fd);
									//cout << "\nsocket conn closed before being serviced" << flush;
									epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
								}
							}
						}
						else
						{
							if(mapo!=NULL)
							{
								tablemap::iterator it;
								long siz = 0;
								if(que[2]=="lt")
								{
									for (it=mapo->begin();it!=mapo->end();it++)
									{
										if(it->first<que[3])
											siz++;
									}
								}
								else if(que[2]=="gt")
								{
									for (it=mapo->begin();it!=mapo->end();it++)
									{
										if(it->first>que[3])
											siz++;
									}
								}
								else if(que[2]=="lf")
								{
									int pos = boost::lexical_cast<int>(que[3]);
									string did = que[4];
									for (it=mapo->begin();it!=mapo->end();it++)
									{
										AMEFObject* obje = decoder.decodeB(it->second,false,true);
										if(obje->getPackets().at(pos)->getValueStr()<did)
										{
											siz++;
										}
										delete obje;
									}
								}
								else if(que[2]=="gf")
								{
									int pos = boost::lexical_cast<int>(que[3]);
									string did = que[4];
									for (it=mapo->begin();it!=mapo->end();it++)
									{
										AMEFObject* obje = decoder.decodeB(it->second,false,true);
										if(obje->getPackets().at(pos)->getValueStr()>did)
										{
											siz++;
										}
										delete obje;
									}
								}
								AMEFObject objn;
								objn.addPacket((long)siz);
								AMEFEncoder encoder;
								string ndata = encoder.encodeWL(&objn, true);
								ndata += "F";
								////cout << dat << endl;
								int err = send(fd, ndata.c_str(), ndata.length(), 0);
								if(err<=0)
								{
									close(events[n].data.fd);
									//cout << "\nsocket conn closed before being serviced" << flush;
									epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
								}
							}
							else
							{
								int err = send(fd, "F", 1, 0);
								if(err<=0)
								{
									close(events[n].data.fd);
									//cout << "\nsocket conn closed before being serviced" << flush;
									epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
								}
							}
						}
					}
					else if(quer.find("select ")!=-1 && quer.find(" = ")!=-1)
					{
						boost::iter_split(que, quer, boost::first_finder(" "));
						tablemap *mapo = cached[que[1]];
						if(mapo!=NULL && mapo->find(que[3])!=mapo->end())
						{
							string dat = (*mapo)[que[3]];
							dat += "F";
							////cout << dat << endl;
							int err = send(fd, dat.c_str(), dat.length(), 0);
							if(err<=0)
							{
								close(events[n].data.fd);
								//cout << "\nsocket conn closed before being serviced" << flush;
								epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
							}
						}
						else
						{
							int err = send(fd, "F", 1, 0);
							if(err<=0)
							{
								close(events[n].data.fd);
								//cout << "\nsocket conn closed before being serviced" << flush;
								epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
							}
						}
					}
					else if(quer.find("selectfkeys ")!=-1 && quer.find(" = ")!=-1)
					{
						boost::iter_split(que, quer, boost::first_finder(" "));
						tablemap *mapo = cached[que[1]];
						string fl;
						for (int var = 3; var < que.size(); ++var) {
							if(mapo->find(que[var])!=mapo->end())
							{
								fl = que[var];
								break;
							}
						}
						if(fl!="")
						{
							string dat = (*mapo)[fl];
							dat += "F";
							////cout << dat << endl;
							int err = send(fd, dat.c_str(), dat.length(), 0);
							if(err<=0)
							{
								close(events[n].data.fd);
								//cout << "\nsocket conn closed before being serviced" << flush;
								epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
							}
						}
						else
						{
							int err = send(fd, "F", 1, 0);
							if(err<=0)
							{
								close(events[n].data.fd);
								//cout << "\nsocket conn closed before being serviced" << flush;
								epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
							}
						}
					}
					else if(quer.find("find ")!=-1)
					{
						vector<string> tbls;
						boost::iter_split(que, quer, boost::first_finder(" "));
						boost::iter_split(tbls, que[1], boost::first_finder(","));
						string flag = "";
						for (int var = 0; var < tbls.size(); var++)
						{
							tablemap *mapo = cached[tbls[var]];
							if(mapo!=NULL)
							{
								if(mapo->find(que[2])!=mapo->end())
								{
									flag = tbls[var];
									break;
								}
							}
						}
						if(flag!="")
						{
							AMEFObject obj;
							obj.addPacket(flag);
							AMEFEncoder enco;
							string dat = enco.encodeWL(&obj,true);
							dat += "F";

							int err = send(fd, dat.c_str(), dat.length(), 0);
							if(err<=0)
							{
								close(events[n].data.fd);
								//cout << "\nsocket conn closed before being serviced" << flush;
								epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
							}
						}
						else
						{
							int err = send(fd, "F", 1, 0);
							if(err<=0)
							{
								close(events[n].data.fd);
								//cout << "\nsocket conn closed before being serviced" << flush;
								epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
							}
						}
					}
					else if(quer.find("selectcnt ")!=-1)
					{
						boost::iter_split(que, quer, boost::first_finder(" "));
						tablemap *mapo = cached[que[1]];
						if(mapo!=NULL)
						{
							tablemap::iterator it;
							string dat;
							long cnter = 1,cnt = 1;
							long st = boost::lexical_cast<long>(que[2]);
							long limit = boost::lexical_cast<long>(que[3]);
							for (it=mapo->begin();it!=mapo->end();it++,cnter++)
							{
								if(cnter>=st)
								{
									dat += it->second;
									cnt++;
								}
								//cout << cnter << endl;
								if(cnt==limit)
									break;
							}
							dat += "F";
							//cout << dat << " " << dat.length()<< endl;
							int totn = dat.length();
							int sentn = 0;
							while(totn>0)
							{
								sentn = send(fd, dat.c_str(), dat.length(), 0);
								//cout << sentn << " " << errno << endl;
								if(sentn>0)
								{
									dat = dat.substr(sentn);
									totn -= sentn;
								}
								else
								{
									usleep(100);
									if(errno==EPIPE)
									{
										close(events[n].data.fd);
										//cout << "\nsocket conn closed before being serviced" << flush;
										epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
										break;

									}
								}
							}
						}
						else
						{
							int err = send(fd, "F", 1, 0);
							if(err<=0)
							{
								close(events[n].data.fd);
								//cout << "\nsocket conn closed before being serviced" << flush;
								epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
							}
						}
					}
					else if(quer.find("select ")!=-1 &&
						(quer.find(" < ")!=-1 ||  quer.find(" > ")!=-1 || quer.find(" *")!=-1
								|| quer.find(" like")!=-1))
					{
						boost::iter_split(que, quer, boost::first_finder(" "));
						tablemap *mapo = cached[que[1]];
						if(mapo!=NULL)
						{
							if(quer.find(" *")!=-1)
							{
								tablemap::iterator it;
								string dat;
								long cnter = 1;
								long limit = mapo->size();
								if(que.size()>4 && que[3]=="limit")
									limit = boost::lexical_cast<long>(que[4]);
								for (it=mapo->begin();it!=mapo->end();it++,cnter++)
								{
									dat += it->second;
									if(cnter==limit)
										break;
								}
								dat += "F";
								int totn = dat.length();
								int sentn = 0;
								while(totn>0)
								{
									sentn = send(fd, dat.c_str(), dat.length(), 0);
									//cout << sentn << " " << errno << endl;
									if(sentn>0)
									{
										dat = dat.substr(sentn);
										totn -= sentn;
									}
									else
									{
										usleep(100);
										if(errno==EPIPE)
										{
											close(events[n].data.fd);
											//cout << "\nsocket conn closed before being serviced" << flush;
											epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
											break;

										}
									}
								}
								//cout << dat << " done sending all" << endl;
							}
							else if(quer.find(" lkbg ")!=-1 && que.size()>=4)
							{
								tablemap::iterator it;
								string dat;
								long cnter = 1;
								long limit = mapo->size();
								if(que[4]=="limit")
									limit = boost::lexical_cast<long>(que[5]);
								for (it=mapo->begin();it!=mapo->end();it++)
								{
									if(it->first.find(que[3])==0)
									{
										dat += it->second;
										if(cnter++>=limit)
											break;
									}
								}
								dat += "F";
								int totn = dat.length();
								int sentn = 0;
								while(totn>0)
								{
									sentn = send(fd, dat.c_str(), dat.length(), 0);
									//cout << sentn << " " << errno << endl;
									if(sentn>0)
									{
										dat = dat.substr(sentn);
										totn -= sentn;
									}
									else
									{
										usleep(100);
										if(errno==EPIPE)
										{
											close(events[n].data.fd);
											//cout << "\nsocket conn closed before being serviced" << flush;
											epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
											break;

										}
									}
								}
								////cout << dat.length() << " " << sentn << " done sending all" << endl;
							}
							else if(quer.find(" lkny ")!=-1 && que.size()>=4)
							{
								tablemap::iterator it;
								string dat;
								long cnter = 1;
								long limit = mapo->size();
								if(que[4]=="limit")
									limit = boost::lexical_cast<long>(que[5]);
								for (it=mapo->begin();it!=mapo->end();it++)
								{
									if(it->first.find(que[3])!=string::npos)
									{
										dat += it->second;
										if(cnter++>=limit)
											break;
									}
								}
								dat += "F";
								int totn = dat.length();
								int sentn = 0;
								while(totn>0)
								{
									sentn = send(fd, dat.c_str(), dat.length(), 0);
									//cout << sentn << " " << errno << endl;
									if(sentn>0)
									{
										dat = dat.substr(sentn);
										totn -= sentn;
									}
									else
									{
										usleep(100);
										if(errno==EPIPE)
										{
											close(events[n].data.fd);
											//cout << "\nsocket conn closed before being serviced" << flush;
											epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
											break;

										}
									}
								}
								////cout << dat.length() << " " << sentn << " done sending all" << endl;
							}
							else if(quer.find(" < ")!=-1 && que.size()>=4)
							{
								tablemap::iterator it;
								string dat;
								long cnter = 1;
								long limit = mapo->size();
								if(que[4]=="limit")
									limit = boost::lexical_cast<long>(que[5]);
								for (it=mapo->begin();it!=mapo->end();it++)
								{
									if(it->first<que[3])
									{
										dat += it->second;
										if(cnter++>=limit)
											break;
									}
								}
								dat += "F";
								int totn = dat.length();
								int sentn = 0;
								while(totn>0)
								{
									sentn = send(fd, dat.c_str(), dat.length(), 0);
									//cout << sentn << " " << errno << endl;
									if(sentn>0)
									{
										dat = dat.substr(sentn);
										totn -= sentn;
									}
									else
									{
										usleep(100);
										if(errno==EPIPE)
										{
											close(events[n].data.fd);
											//cout << "\nsocket conn closed before being serviced" << flush;
											epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
											break;

										}
									}
								}
								////cout << dat.length() << " " << sentn << " done sending all" << endl;
							}
							else if(que.size()>=4)
							{
								tablemap::iterator it;
								string dat;
								long cnter = 1;
								long limit = mapo->size();
								if(que[4]=="limit")
									limit = boost::lexical_cast<long>(que[5]);
								for (it=mapo->begin();it!=mapo->end();it++)
								{
									//cout << it->first << " > " << que[3] << endl;
									if(it->first>que[3])
									{
										dat += it->second;
										if(cnter++>=limit)
											break;
									}
								}
								dat += "F";
								int totn = dat.length();
								int sentn = 0;
								//cout << totn << endl;
								while(totn>0)
								{
									sentn = send(fd, dat.c_str(), dat.length(), 0);
									//cout << sentn << " " << errno << endl;
									if(sentn>0)
									{
										dat = dat.substr(sentn);
										totn -= sentn;
									}
									else
									{
										usleep(100);
										if(errno==EPIPE)
										{
											close(events[n].data.fd);
											//cout << "\nsocket conn closed before being serviced" << flush;
											epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
											break;

										}
									}
								}
							}
							else
							{
								//cout << "invalid command " << quer << endl;
								int err = send(fd, "F", 1, 0);
								if(err<=0)
								{
									close(events[n].data.fd);
									//cout << "\nsocket conn closed before being serviced" << flush;
									epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
								}
							}
						}
						else
						{
							int err = send(fd, "F", 1, 0);
							if(err<=0)
							{
								close(events[n].data.fd);
								//cout << "\nsocket conn closed before being serviced" << flush;
								epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
							}
						}
					}
					else if(quer.find("update ")!=-1)
					{
						bool fl = update(quer, query->getPackets().at(1)->getValue());
						int err = -1;
						if(fl)
							err = send(fd, "T", 1, 0);
						else
							err = send(fd, "F", 1, 0);
						if(err<=0)
						{
							close(events[n].data.fd);
							//cout << "\nsocket conn closed before being serviced" << flush;
							epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						}
					}
					else if(quer.find("updmov ")!=-1)
					{
						string val = query->getPackets().at(1)->getValue();
						bool fl = updmov(quer,val,true);
						int err = -1;
						if(fl)
							err = send(fd, "T", 1, 0);
						else
							err = send(fd, "F", 1, 0);
						if(err<=0)
						{
							close(events[n].data.fd);
							//cout << "\nsocket conn closed before being serviced" << flush;
							epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						}
					}
					else if(quer.find("updcpy ")!=-1)
					{
						string val = query->getPackets().at(1)->getValue();
						bool fl = updmov(quer,val,false);
						int err = -1;
						if(fl)
							err = send(fd, "T", 1, 0);
						else
							err = send(fd, "F", 1, 0);
						if(err<=0)
						{
							close(events[n].data.fd);
							//cout << "\nsocket conn closed before being serviced" << flush;
							epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						}
					}
					else if(quer.find("upmovwrt ")!=-1)
					{
						string val = query->getPackets().at(1)->getValue();
						bool fl = upmovwrt(quer,val,true);
						int err = -1;
						if(fl)
							err = send(fd, "T", 1, 0);
						else
							err = send(fd, "F", 1, 0);
						if(err<=0)
						{
							close(events[n].data.fd);
							//cout << "\nsocket conn closed before being serviced" << flush;
							epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						}
					}
					else if(quer.find("upcpywrt ")!=-1)
					{
						string val = query->getPackets().at(1)->getValue();
						bool fl = upmovwrt(quer,val,false);
						int err = -1;
						if(fl)
							err = send(fd, "T", 1, 0);
						else
							err = send(fd, "F", 1, 0);
						if(err<=0)
						{
							close(events[n].data.fd);
							//cout << "\nsocket conn closed before being serviced" << flush;
							epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						}
					}
					else if(quer.find("upmovwrtpvap ")!=-1)
					{
						string val = query->getPackets().at(1)->getValue();
						bool fl = upmovwrtpvap(quer,val,true);
						int err = -1;
						if(fl)
							err = send(fd, "T", 1, 0);
						else
							err = send(fd, "F", 1, 0);
						if(err<=0)
						{
							close(events[n].data.fd);
							//cout << "\nsocket conn closed before being serviced" << flush;
							epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						}
					}
					else if(quer.find("upcpywrtpvap ")!=-1)
					{
						string val = query->getPackets().at(1)->getValue();
						bool fl = upmovwrtpvap(quer,val,false);
						int err = -1;
						if(fl)
							err = send(fd, "T", 1, 0);
						else
							err = send(fd, "F", 1, 0);
						if(err<=0)
						{
							close(events[n].data.fd);
							//cout << "\nsocket conn closed before being serviced" << flush;
							epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						}
					}
					/*else if(quer.find("remfldmv ")!=-1 && quer.find(" = ")!=-1)
					{
						boost::iter_split(que, quer, boost::first_finder(" "));
						tablemap *mapo = cached[que[1]];
						tablemap *mapt = cached[que[2]];
						if(mapo!=NULL)
						{
							int fld = boost::lexical_cast<int>(que[4]);
							string dat = (*mapo)[que[3]];
							AMEFObject* objo = decoder.decodeB(dat,false,true);
							objo->getPackets().erase(objo->getPackets().begin()+fld);
							AMEFEncoder encoder;
							string ndata = encoder.encodeWL(objo, true);
							lock.lock();
							mapo->erase(key);
							mapo->insert(pair<string,string>(que[3], ndata));
							lock.unlock();
							dat += "T";
							////cout << dat << endl;
							int err = send(fd, dat.c_str(), dat.length(), 0);
							if(err<=0)break;
						}
						else
						{
							int err = send(fd, "F", 1, 0);
							if(err<=0)break;
						}
					}*/
					else if(quer.find("move ")!=-1)
					{
						bool fl = move(quer,true);
						int err = -1;
						if(fl)
							err = send(fd, "T", 1, 0);
						else
							err = send(fd, "F", 1, 0);
						if(err<=0)
						{
							close(events[n].data.fd);
							//cout << "\nsocket conn closed before being serviced" << flush;
							epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						}
					}
					else if(quer.find("copy ")!=-1)
					{
						bool fl = move(quer,false);
						int err = -1;
						if(fl)
							err = send(fd, "T", 1, 0);
						else
							err = send(fd, "F", 1, 0);
						if(err<=0)
						{
							close(events[n].data.fd);
							//cout << "\nsocket conn closed before being serviced" << flush;
							epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						}
					}
					else if(quer.find("movepvap ")!=-1)//move to a table name derived from the to table and the property at index of the from table
					{
						bool fl = movepvap(quer,true);
						int err = -1;
						if(fl)
							err = send(fd, "T", 1, 0);
						else
							err = send(fd, "F", 1, 0);
						if(err<=0)
						{
							close(events[n].data.fd);
							//cout << "\nsocket conn closed before being serviced" << flush;
							epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						}
					}
					else if(quer.find("copypvap ")!=-1)//move to a table name derived from the to table and the property at index of the from table
					{
						bool fl = movepvap(quer,false);
						int err = -1;
						if(fl)
							err = send(fd, "T", 1, 0);
						else
							err = send(fd, "F", 1, 0);
						if(err<=0)
						{
							close(events[n].data.fd);
							//cout << "\nsocket conn closed before being serviced" << flush;
							epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						}
					}
					else if(quer.find("clrtmp")!=-1)//Not thread safe
					{
						tablemap *mapt = cached["temp"];
						mapt->empty();
						mapt->clear();
						delete mapt;
						cached["temp"] = new map<string, string>();
						int err = send(fd, "T", 1, 0);
						if(err<=0)
						{
							close(events[n].data.fd);
							//cout << "\nsocket conn closed before being serviced" << flush;
							epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						}
					}
					else if(quer.find("mvtmp ")!=-1)//Not thread safe
					{
						tablemap::iterator it;
						boost::iter_split(que, quer, boost::first_finder(" "));
						tablemap *mapf = cached[que[1]];

						if(mapf==NULL)
						{
							int err = send(fd, "F", 1, 0);
							if(err<=0)
							{
								close(events[n].data.fd);
								//cout << "\nsocket conn closed before being serviced" << flush;
								epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
							}
						}
						else
						{
							tablemap *mapt = cached["temp"];
							mapt->empty();
							mapt->clear();
							delete mapt;
							mapt = new map<string, string>(*mapf);
							lock.lock();
							mapf->empty();
							mapf->clear();
							lock.unlock();

							int err = send(fd, "T", 1, 0);
							if(err<=0)
							{
								close(events[n].data.fd);
								//cout << "\nsocket conn closed before being serviced" << flush;
								epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
							}
						}
					}
					else if(quer.find("movegt ")!=-1)
					{
						tablemap::iterator it;
						boost::iter_split(que, quer, boost::first_finder(" "));
						tablemap *mapf = cached[que[1]];
						tablemap *mapt = cached[que[2]];
						long cnter = 1;
						long limit = mapf->size();

						if(que.size()==4 || (que.size()>4 && que[4]=="limit"))
						{
							if(que.size()>4 && que[4]=="limit")
								limit = boost::lexical_cast<long>(que[5]);
							if(mapf==NULL || mapt==NULL)
							{
								int err = send(fd, "F", 1, 0);
								if(err<=0)
								{
									close(events[n].data.fd);
									//cout << "\nsocket conn closed before being serviced" << flush;
									epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
								}
							}
							else
							{
								string did = que[3];
								for (it=mapf->begin();it!=mapf->end();it++)
								{
									if(it->first>did)
									{
										lock.lock();
										mapt->insert(pair<string,string>(it->first, (*mapf)[it->first]));
										mapf->erase(it->first);
										lock.unlock();
										if(cnter++>limit)
											break;
									}
								}
								int err = send(fd, "T", 1, 0);
								if(err<=0)
								{
									close(events[n].data.fd);
									//cout << "\nsocket conn closed before being serviced" << flush;
									epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
								}
							}
						}
						else
						{
							if(que.size()>5 && que[6]=="limit")
								limit = boost::lexical_cast<long>(que[7]);
							if(mapf==NULL || mapt==NULL)
							{
								int err = send(fd, "F", 1, 0);
								if(err<=0)
								{
									close(events[n].data.fd);
									//cout << "\nsocket conn closed before being serviced" << flush;
									epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
								}
							}
							else
							{
								int pos = boost::lexical_cast<int>(que[3]);
								string did = que[4];
								for (it=mapf->begin();it!=mapf->end();it++)
								{
									AMEFObject* obje = decoder.decodeB(it->second,false,true);
									if(obje->getPackets().at(pos)->getValueStr()>did)
									{
										lock.lock();
										mapt->insert(pair<string,string>(it->first, (*mapf)[it->first]));
										mapf->erase(it->first);
										lock.unlock();
										if(cnter++>limit)
											break;
									}
									delete obje;
								}
								int err = send(fd, "T", 1, 0);
								if(err<=0)
								{
									close(events[n].data.fd);
									//cout << "\nsocket conn closed before being serviced" << flush;
									epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
								}
							}
						}
					}
					else if(quer.find("movelt ")!=-1)
					{
						tablemap::iterator it;
						boost::iter_split(que, quer, boost::first_finder(" "));
						tablemap *mapf = cached[que[1]];
						tablemap *mapt = cached[que[2]];
						long cnter = 1;
						long limit = mapf->size();

						if(que.size()==4 || (que.size()>4 && que[4]=="limit"))
						{
							if(que.size()>4 && que[4]=="limit")
								limit = boost::lexical_cast<long>(que[5]);
							if(mapf==NULL || mapt==NULL)
							{
								int err = send(fd, "F", 1, 0);
								if(err<=0)
								{
									close(events[n].data.fd);
									//cout << "\nsocket conn closed before being serviced" << flush;
									epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
								}
							}
							else
							{
								string did = que[3];
								for (it=mapf->begin();it!=mapf->end();it++)
								{
									if(it->first<did)
									{
										lock.lock();
										mapt->insert(pair<string,string>(it->first, (*mapf)[it->first]));
										mapf->erase(it->first);
										lock.unlock();
										if(cnter++>limit)
											break;
									}
								}
								int err = send(fd, "T", 1, 0);
								if(err<=0)
								{
									close(events[n].data.fd);
									//cout << "\nsocket conn closed before being serviced" << flush;
									epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
								}
							}
						}
						else
						{
							if(que.size()>5 && que[6]=="limit")
								limit = boost::lexical_cast<long>(que[7]);
							if(mapf==NULL || mapt==NULL)
							{
								int err = send(fd, "F", 1, 0);
								if(err<=0)
								{
									close(events[n].data.fd);
									//cout << "\nsocket conn closed before being serviced" << flush;
									epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
								}
							}
							else
							{
								int pos = boost::lexical_cast<int>(que[3]);
								string did = que[4];
								for (it=mapf->begin();it!=mapf->end();it++)
								{
									AMEFObject* obje = decoder.decodeB(it->second,false,true);
									if(obje->getPackets().at(pos)->getValueStr()<did)
									{
										lock.lock();
										mapt->insert(pair<string,string>(it->first, (*mapf)[it->first]));
										mapf->erase(it->first);
										lock.unlock();
										if(cnter++>limit)
											break;
									}
									delete obje;
								}
								int err = send(fd, "T", 1, 0);
								if(err<=0)
								{
									close(events[n].data.fd);
									//cout << "\nsocket conn closed before being serviced" << flush;
									epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
								}
							}
						}
					}
					else if(quer.find("movewrt ")!=-1)
					{
						bool fl = movewrt(quer,true);
						int err = -1;
						if(fl)
							err = send(fd, "T", 1, 0);
						else
							err = send(fd, "F", 1, 0);
						if(err<=0)
						{
							close(events[n].data.fd);
							//cout << "\nsocket conn closed before being serviced" << flush;
							epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						}
					}
					else if(quer.find("copywrt ")!=-1)
					{
						bool fl = movewrt(quer,false);
						int err = -1;
						if(fl)
							err = send(fd, "T", 1, 0);
						else
							err = send(fd, "F", 1, 0);
						if(err<=0)
						{
							close(events[n].data.fd);
							//cout << "\nsocket conn closed before being serviced" << flush;
							epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						}
					}
					else if(quer.find("movewrtpvap ")!=-1)
					{
						bool fl = movewrtpvap(quer,true);
						int err = -1;
						if(fl)
							err = send(fd, "T", 1, 0);
						else
							err = send(fd, "F", 1, 0);
						if(err<=0)
						{
							close(events[n].data.fd);
							//cout << "\nsocket conn closed before being serviced" << flush;
							epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						}
					}
					else if(quer.find("movewrtpvap ")!=-1)
					{
						bool fl = movewrtpvap(quer,false);
						int err = -1;
						if(fl)
							err = send(fd, "T", 1, 0);
						else
							err = send(fd, "F", 1, 0);
						if(err<=0)
						{
							close(events[n].data.fd);
							//cout << "\nsocket conn closed before being serviced" << flush;
							epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						}
					}
					else if(quer.find("apmov ")!=-1)
					{
						string val = query->getPackets().at(1)->getValue();
						bool fl = apmov(quer,val,true);
						int err = -1;
						if(fl)
							err = send(fd, "T", 1, 0);
						else
							err = send(fd, "F", 1, 0);
						if(err<=0)
						{
							close(events[n].data.fd);
							//cout << "\nsocket conn closed before being serviced" << flush;
							epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						}
					}
					else if(quer.find("apcpy ")!=-1)
					{
						string val = query->getPackets().at(1)->getValue();
						bool fl = apmov(quer,val,false);
						int err = -1;
						if(fl)
							err = send(fd, "T", 1, 0);
						else
							err = send(fd, "F", 1, 0);
						if(err<=0)
						{
							close(events[n].data.fd);
							//cout << "\nsocket conn closed before being serviced" << flush;
							epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						}
					}
					else if(quer.find("apmovwrt ")!=-1)
					{
						string val = query->getPackets().at(1)->getValue();
						bool fl = apmovwrt(quer,val,true);
						int err = -1;
						if(fl)
							err = send(fd, "T", 1, 0);
						else
							err = send(fd, "F", 1, 0);
						if(err<=0)
						{
							close(events[n].data.fd);
							//cout << "\nsocket conn closed before being serviced" << flush;
							epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						}
					}
					else if(quer.find("apcpywrt ")!=-1)
					{
						string val = query->getPackets().at(1)->getValue();
						bool fl = apmovwrt(quer,val,false);
						int err = -1;
						if(fl)
							err = send(fd, "T", 1, 0);
						else
							err = send(fd, "F", 1, 0);
						if(err<=0)
						{
							close(events[n].data.fd);
							//cout << "\nsocket conn closed before being serviced" << flush;
							epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						}
					}
					else if(quer.find("apmovwrtpvap ")!=-1)
					{
						string val = query->getPackets().at(1)->getValue();
						bool fl = apmovwrtpvap(quer,val,true);
						int err = -1;
						if(fl)
							err = send(fd, "T", 1, 0);
						else
							err = send(fd, "F", 1, 0);
						if(err<=0)
						{
							close(events[n].data.fd);
							//cout << "\nsocket conn closed before being serviced" << flush;
							epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						}
					}
					else if(quer.find("apcpywrtpvap ")!=-1)
					{
						string val = query->getPackets().at(1)->getValue();
						bool fl = apmovwrtpvap(quer,val,false);
						int err = -1;
						if(fl)
							err = send(fd, "T", 1, 0);
						else
							err = send(fd, "F", 1, 0);
						if(err<=0)
						{
							close(events[n].data.fd);
							//cout << "\nsocket conn closed before being serviced" << flush;
							epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						}
					}
					else if(quer.find("delete ")!=-1)
					{
						////cout << quer << endl;
						boost::iter_split(que, quer, boost::first_finder(" "));
						tablemap *mapo = cached[que[1]];
						if(mapo==NULL)
						{
							int err = send(fd, "F", 1, 0);
							if(err<=0)
							{
								close(events[n].data.fd);
								//cout << "\nsocket conn closed before being serviced" << flush;
								epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
							}
						}
						else
						{
							////cout << que[1] << mapo->size() << que[2]<<  endl;
							if(que[2]=="*")
							{
								lock.lock();
								mapo->empty();
								mapo->clear();
								lock.unlock();
							}
							else if(que[2]=="=")
							{
								lock.lock();
								mapo->erase(que[3]);
								lock.unlock();
							}
							else if(que[2]=="<")
							{
								tablemap::iterator it;
								string did = que[3];
								for (it=mapo->begin();it!=mapo->end();it++)
								{
									if(it->first<did)
									{
										lock.lock();
										mapo->erase(it->first);
										lock.unlock();
									}
								}
							}
							else if(que[2]==">")
							{
								tablemap::iterator it;
								string did = que[3];
								for (it=mapo->begin();it!=mapo->end();it++)
								{
									if(it->first>did)
									{
										lock.lock();
										mapo->erase(it->first);
										lock.unlock();
									}
								}
							}
							else if(que[2]=="lf")
							{
								tablemap::iterator it;
								int pos = boost::lexical_cast<int>(que[3]);
								string did = que[4];
								for (it=mapo->begin();it!=mapo->end();it++)
								{
									AMEFObject* obje = decoder.decodeB(it->second,false,true);
									if(obje->getPackets().at(pos)->getValueStr()<did)
									{
										lock.lock();
										mapo->erase(it->first);
										lock.unlock();
									}
									delete obje;
								}
							}
							else if(que[2]=="gf")
							{
								tablemap::iterator it;
								int pos = boost::lexical_cast<int>(que[3]);
								string did = que[4];
								for (it=mapo->begin();it!=mapo->end();it++)
								{
									AMEFObject* obje = decoder.decodeB(it->second,false,true);
									if(obje->getPackets().at(pos)->getValueStr()>did)
									{
										lock.lock();
										mapo->erase(it->first);
										lock.unlock();
									}
									delete obje;
								}
							}
							int err = send(fd, "T", 1, 0);
							if(err<=0)
							{
								close(events[n].data.fd);
								//cout << "\nsocket conn closed before being serviced" << flush;
								epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
							}
						}
					}
					else if(quer.find("getkeys ")!=-1)
					{
						boost::iter_split(que, quer, boost::first_finder(" "));
						tablemap *mapo = cached[que[1]];
						if(mapo!=NULL)
						{
							tablemap::iterator it;
							AMEFObject obj;
							for (it=mapo->begin();it!=mapo->end();it++)
							{
								obj.addPacket(it->first);
							}
							AMEFEncoder enco;
							string dat = enco.encodeWL(&obj,true);
							dat += "F";
							//cout << dat << endl;
							int totn = dat.length();
							int sentn = 0;
							while(totn>0)
							{
								sentn = send(fd, dat.c_str(), dat.length(), 0);
								//cout << sentn << " " << errno << endl;
								if(sentn>0)
								{
									dat = dat.substr(sentn);
									totn -= sentn;
								}
								else
								{
									usleep(100);
									if(errno==EPIPE)
									{
										close(events[n].data.fd);
										//cout << "\nsocket conn closed before being serviced" << flush;
										epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
										break;

									}
								}
							}
						}
						else
						{
							int err = send(fd, "F", 1, 0);
							if(err<=0)
							{
								close(events[n].data.fd);
								//cout << "\nsocket conn closed before being serviced" << flush;
								epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
							}
						}
					}
					else if(quer.find("add2nfld ")!=-1)
					{
						string val = query->getPackets().at(1)->getValue();
						bool fl = updatefld(quer,val,true);
						int err = -1;
						if(fl)
							err = send(fd, "T", 1, 0);
						else
							err = send(fd, "F", 1, 0);
						if(err<=0)
						{
							close(events[n].data.fd);
							//cout << "\nsocket conn closed before being serviced" << flush;
							epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						}
					}
					else if(quer.find("sub2nfld ")!=-1)
					{
						string val = query->getPackets().at(1)->getValue();
						bool fl = updatefld(quer,val,false);
						int err = -1;
						if(fl)
							err = send(fd, "T", 1, 0);
						else
							err = send(fd, "F", 1, 0);
						if(err<=0)
						{
							close(events[n].data.fd);
							//cout << "\nsocket conn closed before being serviced" << flush;
							epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
						}
					}
					else if(quer.find("getkeypairs ")!=-1)
					{
						boost::iter_split(que, quer, boost::first_finder(" "));
						tablemap *mapo = cached[que[1]];
						if(mapo!=NULL)
						{
							tablemap::iterator it;

							long cnter = 0;
							long st = boost::lexical_cast<long>(que[2]);
							long limit = boost::lexical_cast<long>(que[3]);
							long cnt = 0;
							AMEFEncoder enco;
							string dat;

							for (it=mapo->begin();it!=mapo->end();it++,cnter++)
							{
								if(cnter>=st)
								{
									AMEFObject obj;
									cnt++;
									obj.addPacket(it->first);
									obj.addPacket(it->second);
									dat += enco.encodeWL(&obj,true);
									if(cnt>limit)
										break;
								}
							}

							dat += "F";

							//cout << dat << endl;
							int totn = dat.length();
							int sentn = 0;
							while(totn>0)
							{
								sentn = send(fd, dat.c_str(), dat.length(), 0);
								//cout << sentn << " " << errno << endl;
								if(sentn>0)
								{
									dat = dat.substr(sentn);
									totn -= sentn;
								}
								else
								{
									usleep(100);
									if(errno==EPIPE)
									{
										close(events[n].data.fd);
										//cout << "\nsocket conn closed before being serviced" << flush;
										epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
										break;

									}
								}
							}
						}
						else
						{
							int err = send(fd, "F", 1, 0);
							if(err<=0)
							{
								close(events[n].data.fd);
								//cout << "\nsocket conn closed before being serviced" << flush;
								epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
							}
						}
					}
					else if(quer.find("getkplk ")!=-1)
					{
						boost::iter_split(que, quer, boost::first_finder(" "));
						tablemap *mapo = cached[que[1]];
						if(mapo!=NULL)
						{
							tablemap::iterator it;

							long cnter = 0;
							//long limit = boost::lexical_cast<long>(que[3]);
							long cnt = 0;
							AMEFEncoder enco;
							string dat;

							for (it=mapo->begin();it!=mapo->end();it++)
							{
								if(it->first.find(que[2])==0)
								{
									AMEFObject obj;
									obj.addPacket(it->first);
									obj.addPacket(it->second);
									dat += enco.encodeWL(&obj,true);
								}
							}

							dat += "F";

							//cout << dat << endl;
							int totn = dat.length();
							int sentn = 0;
							while(totn>0)
							{
								sentn = send(fd, dat.c_str(), dat.length(), 0);
								//cout << sentn << " " << errno << endl;
								if(sentn>0)
								{
									dat = dat.substr(sentn);
									totn -= sentn;
								}
								else
								{
									usleep(100);
									if(errno==EPIPE)
									{
										close(events[n].data.fd);
										//cout << "\nsocket conn closed before being serviced" << flush;
										epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
										break;

									}
								}
							}
						}
						else
						{
							int err = send(fd, "F", 1, 0);
							if(err<=0)
							{
								close(events[n].data.fd);
								//cout << "\nsocket conn closed before being serviced" << flush;
								epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
							}
						}
					}
					//query->~AMEFObject();
					//cout << "end=" << query << endl;
					delete query;
					usleep(100);
					quer.clear();
					data.clear();
					ln.clear();
					que.clear();
					memset(&buf[0], 0, sizeof(buf));
					//cout << "endnull=" << query << endl;
				}
				else
				{
					close(events[n].data.fd);
					cout << "WARNING>> Invalid data or someone trying to attack the cache or maybe a simple telnet attempt...ignore..." << flush;
					epoll_ctl(epoll_handle, EPOLL_CTL_DEL, events[n].data.fd,&ev);
				}
			}
		}
	}
    ifs.close();
    backupTables();
    return 0;
}

void bootFromBackupTables()
{
	PropFileReader prop;
	propMap props = prop.getProperties("gcache.prop");

	string tables = props["tables"];

	strVec ipprts;
	boost::iter_split(ipprts, tables, boost::first_finder(","));
	for (int var = 0; var < (int)ipprts.size(); var++)
	{
		tablemap *mapo = cached[ipprts.at(var)];
		tablemap::iterator it;
		ifstream ifskeys((ipprts.at(var)+"_keys.dat").c_str(), std::ios::in|std::ios::binary);
		ifstream ifsdata((ipprts.at(var)+"_data.dat").c_str(), std::ios::in|std::ios::binary);
		strVec que;

		if (ifskeys.is_open() && ifskeys.is_open())
		{
			ifskeys.seekg (0, ios::end);
			int length = ifskeys.tellg();
			if(length==0)
				continue;
			ifskeys.seekg (0, ios::beg);
			cout << "Loading table " << ipprts.at(var) << " from backup files " << endl;
			string temp;
			while(getline(ifskeys, temp))
			{
				boost::iter_split(que, temp, boost::first_finder(","));
				int len = boost::lexical_cast<int>(que[1]);
				char *data = new char[len];
				ifsdata.read(data,len);
				string value;
				for (int var = 0; var < len; ++var) {
					value.push_back(data[var]);
				}
				(*mapo)[que[0]] = value;
				delete data;
				cout << que[0] << " = "<< value << endl;
			}
			cout << "Completed load from backup file for table " << ipprts.at(var) << endl;
		}
		ifskeys.close();
		ifsdata.close();

	}
}


void backupTables()
{
	PropFileReader prop;
	propMap props = prop.getProperties("gcache.prop");

	string tables = props["tables"];

	strVec ipprts;
	boost::iter_split(ipprts, tables, boost::first_finder(","));
	for (int var = 0; var < (int)ipprts.size(); var++)
	{
		tablemap *mapo = cached[ipprts.at(var)];
		if(mapo==NULL || mapo->size()==0)
			continue;
		cout << "Starting backup of table " << ipprts.at(var) << endl;
		tablemap::iterator it;
		ofstream ofskeys((ipprts.at(var)+"_keys.dat").c_str(), std::ios::binary|std::ios::out);
		ofstream ofsdata((ipprts.at(var)+"_data.dat").c_str(), std::ios::binary|std::ios::out);
		for (it=mapo->begin();it!=mapo->end();it++)
		{
			string value = it->second;
			string key = it->first + "," + boost::lexical_cast<string>(value.length()) + "\n";
			ofskeys.write(key.c_str(),key.length());
			for (int var = 0; var < (int)value.length(); ++var) {
				ofsdata.put(value[var]);
			}
			cout << key << " = "<< value << endl;
		}
		ofskeys.close();
		ofsdata.close();
		cout << "Created backup files for table " << ipprts.at(var) << " complete" << endl;
	}
}

void static backupIntermediateTables()//To be worked upon.....
{
	while(true)
	{
		backupTables();
		sleep(60);
	}
}

